package com.bangcommunity.bbframe.fdfs;

import com.bangcommunity.bbframe.fdfs.connection.StorageConnection;
import com.bangcommunity.bbframe.fdfs.connection.TrackerConnection;
import org.apache.commons.pool2.impl.GenericKeyedObjectPool;
import org.csource.fastdfs.TrackerGroup;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.concurrent.atomic.AtomicInteger;

public class FdfsClientTemplate {

    private static Logger logger = LoggerFactory.getLogger(FdfsClientTemplate.class);
    private GenericKeyedObjectPool<String, TrackerConnection> trackerClientPool;
    private GenericKeyedObjectPool<StorageAddress, StorageConnection> storageClientPool;
    private TrackerGroup gTrackerGroup;

    public FdfsClientTemplate(TrackerGroup gTrackerGroup,
            GenericKeyedObjectPool<String, TrackerConnection> trackerClientPool,
            GenericKeyedObjectPool<StorageAddress, StorageConnection> storageClientPool) {
        super();
        this.gTrackerGroup = gTrackerGroup;
        this.trackerClientPool = trackerClientPool;
        this.storageClientPool = storageClientPool;
    }

    public void close() {
        this.trackerClientPool.close();
        this.storageClientPool.close();
    }

    public <T> T execute(String group, String fileId, StorageAddressInnerFecher storageConnectionInnerFecher,
            FdfsClientExecutor<T> executor) throws Exception {
        String trackerAddr = getTrackerAddr();
        TrackerConnection trackerConnection = null;
        StorageConnection storageConnection = null;
        StorageAddress storageAddr = null;
        try {
            trackerConnection = trackerClientPool.borrowObject(trackerAddr);
            storageAddr = storageConnectionInnerFecher.fetchStorageAddress(trackerConnection);
            storageConnection = storageClientPool.borrowObject(storageAddr);
            return executor.doInExecutor(group, fileId, storageConnection);
        } catch (Exception e) {
            logger.error(e.getMessage());
            throw e;
        } finally {
            if (storageConnection != null) {
                storageClientPool.returnObject(storageAddr, storageConnection);
            }
            if (trackerConnection != null) {
                trackerClientPool.returnObject(trackerAddr, trackerConnection);
            }
        }
    }

    @FunctionalInterface
    public interface StorageAddressInnerFecher {
        StorageAddress fetchStorageAddress(TrackerConnection trackerConnection) throws Exception;
    }

    public <T> T executeReadable(String fullFileId, FdfsClientExecutor<T> executor) throws Exception {
        String[] fullFileParts = FdfsClientUtil.splitFileId(fullFileId);
        String group = fullFileParts[0];
        String fileId = fullFileParts[1];
        return executeReadable(group, fileId, executor);
    }

    public <T> T executeReadable(String group, String fileId, FdfsClientExecutor<T> executor) throws Exception {
        return execute(group, fileId, (trackerConnection) -> trackerConnection.getReadableStorage(group, fileId),
                executor);
    }

    public <T> T executeWritable(String group, FdfsClientExecutor<T> executor) throws Exception {
        return execute(group, null, (trackerConnection) -> trackerConnection.getWritableStorage(group), executor);
    }

    public <T> T executeUpdatable(String fullFileId, FdfsClientExecutor<T> executor) throws Exception {
        String[] fullFileParts = FdfsClientUtil.splitFileId(fullFileId);
        String group = fullFileParts[0];
        String fileId = fullFileParts[1];
        return executeUpdatable(group, fileId, executor);
    }

    public <T> T executeUpdatable(String group, String fileId, FdfsClientExecutor<T> executor) throws Exception {
        return execute(group, fileId, (trackerConnection) -> trackerConnection.getUpdatableStorage(group, fileId),
                executor);
    }

    private AtomicInteger atomicInteger = new AtomicInteger(0);

    public String getTrackerAddr() throws IOException {
        int index = atomicInteger.getAndIncrement();
        if (index > 10000000) {
            atomicInteger.set(0);
        }
        InetSocketAddress[] trackerServers = gTrackerGroup.tracker_servers;
        index %= trackerServers.length;
        InetSocketAddress trackerServer = trackerServers[index];
        return String.valueOf(trackerServer.getHostString() + ":" + trackerServer.getPort());
    }

}
